Greptile Summary

This PR introduces expansion-aware dynamic batching for the explode operator to prevent memory pressure from high-cardinality expansions. The implementation monitors the explode's expansion ratio (output rows / input rows) and dynamically reduces upstream batch size requirements proportionally, so that when an explode has 100x expansion, it requests ~100x fewer rows from upstream to produce appropriately sized downstream batches.

Key changes:

• Added ExpansionAwareBatchingStrategy with exponential moving average (EMA) smoothing to handle varying expansion ratios gracefully
• Modified IntermediateOperator::batching_strategy trait method signature to accept morsel_size_requirement as a parameter instead of operators calling self.morsel_size_requirement() internally
• Added as_any() method to RuntimeStats trait to enable downcasting for type-specific stats access in batching strategies
• Updated all intermediate operators (filter, project, udf, distributed_actor_pool_project, into_batches, unpivot) to match the new trait signature
• Comprehensive test coverage for the new batching strategy including edge cases for high/low expansion, smoothing, and cumulative stats

The implementation includes safety bounds (MIN_EXPANSION, MAX_REDUCTION) to prevent pathological cases and uses cumulative statistics from Counter metrics.

Confidence Score: 5/5

• This PR is safe to merge with minimal risk
• The implementation is well-tested with comprehensive unit tests covering edge cases, uses proper safety bounds to prevent pathological behavior, and follows the established patterns from the filter operator's dynamic batching (PR feat(dynamic batching): add new selectivity aware batching strategy for filter op #5924). The trait signature change is consistently applied across all operators, and the cumulative stats handling is correct.
• No files require special attention

Important Files Changed

Sequence Diagram

sequenceDiagram
    participant Upstream as Upstream Operator
    participant BE as BatchManager
    participant EO as ExplodeOperator
    participant ES as ExpansionState
    participant Stats as ExplodeStats
    participant Downstream as Downstream Operator

    Note over BE: Initial requirement from<br/>downstream (e.g., Strict(1000))
    BE->>Upstream: Request initial batch<br/>(Strict(1000) rows)
    Upstream->>EO: Send 1000 rows
    EO->>EO: Explode operation<br/>(e.g., 100x expansion)
    EO->>Stats: Update rows_in=1000,<br/>rows_out=100000
    EO->>Downstream: Send 100000 rows
    
    Note over BE,ES: Record execution stats
    BE->>ES: record_execution_stat(stats)
    ES->>Stats: Load cumulative counters<br/>(rows_in, rows_out)
    ES->>ES: Calculate expansion ratio<br/>(100000/1000 = 100)
    ES->>ES: Apply EMA smoothing<br/>smoothed_expansion = 100
    
    Note over BE: Calculate new requirements
    BE->>ES: calculate_new_requirements()
    ES->>ES: reduction = 1/100 = 0.01
    ES->>BE: Return Strict(10)
    
    Note over BE: Next batch uses reduced size
    BE->>Upstream: Request reduced batch<br/>(Strict(10) rows)
    Upstream->>EO: Send 10 rows
    EO->>EO: Explode operation<br/>(~100x expansion)
    EO->>Stats: Update rows_in=1010,<br/>rows_out=101000
    EO->>Downstream: Send ~1000 rows
    
    Note over Downstream: Receives appropriately<br/>sized batches

Codecov Report

❌ Patch coverage is 86.25954% with 36 lines in your changes missing coverage. Please review.
✅ Project coverage is 72.93%. Comparing base (7bec778) to head (d8787e6).
⚠️ Report is 290 commits behind head on main.

@@            Coverage Diff             @@
##             main    #6069      +/-   ##
==========================================
+ Coverage   72.91%   72.93%   +0.01%     
==========================================
  Files         973      973              
  Lines      126166   126407     +241     
==========================================
+ Hits        91995    92191     +196     
- Misses      34171    34216      +45     

@universalmind303 are you still working on this?

closing for now. Will try to revisit with the proactive approach as suggested by @colin-ho.